package com.xxx.flink;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 数据流处理
 */
public class StreamWordCount {

    public static void main(String[] args) throws Exception {
        // 创建数据流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);  // 设置任务并行执行数

        // 读取有界数据流
//        String filePath = "input/hello.txt";
//        DataStream<String> inputDataStream = env.readTextFile(filePath);

        // 读取流数据
        // hostname和port可以从命令行参数获取： --host=localhost --port=7777
//        ParameterTool parameterTool = ParameterTool.fromArgs(args);
//        String host = parameterTool.get("host");
//        int port = parameterTool.getInt("port");
        // 读取数据流，这里从socket读取数据流（在命令窗口里执行： nc -l 7777）
        DataStream<String> inputDataStream = env.socketTextStream("localhost", 7777);
        inputDataStream.print("nc");

        // 基于数据流进行数据转换
        DataStream<Tuple2<String,Integer>> resultStream = inputDataStream.flatMap(new WordCount.MyFlatMapper())
                .keyBy(e -> e.f0)
                .sum(1)
                .setParallelism(2);

        resultStream.print().setParallelism(1);

        // 以上属于定义任务，下面是提交任务并执行
        env.execute();
    }

}
